这个名字|媒介_SpringCloudStream消息驱动设计思想以及整合rabbitmq消息队列案例学习笔记

篇首语：本文由编程笔记#小编为大家整理，主要介绍了SpringCloud Stream消息驱动设计思想以及整合rabbitmq消息队列案例--学习笔记相关的知识，希望对你有一定的参考价值。

一&＃xff0c;RabbitMQ的安装和配置 并启动

参见 RabbitMQ的安装和配置https://blog.csdn.net/weixin_43025151/article/details/123186641
RabbitMQ启动成功&＃xff1a;

二&＃xff0c;SpringCloud Stream消息驱动

1&＃xff0c;设计思想

一个标准的消息队列MQ,如下图&＃xff1a;

为什么用CloudStream&＃xff1f;


一句话&＃xff1a;CloudStream屏蔽底层消息中间件的差异&＃xff0c;降低切换成本&＃xff0c;统一消息的编程模型&＃xff01;

通过定义绑定器Binder作为中间件&＃xff0c;实现了应用程序与消息中间件细节之间的隔离。

绑定器Binder&＃xff1a;INPUT适用于消费者 OUTPUT适用于生产者
Stream中的消息通讯方式遵循了发布-订阅模式 Topic主题进行广播 在RabbitMQ就是Exchange 在Kafka中就是Topic。

2&＃xff0c;Springcloud Stream标准流程套路

Binder&＃xff1a;很方便的连接中间件&＃xff0c;屏蔽差异

Channel&＃xff1a;通道&＃xff0c;是队列Queue的一种抽象&＃xff0c;在消息通讯系统中就是实现存储和转发的媒介&＃xff0c;通过channel对队列进行配置

Source和Sink&＃xff1a;简单的可理解为参照对象是Springcloud

Stream自身&＃xff0c;从Stream发布消息就是输出&＃xff0c;接受消息就是输入

3&＃xff0c;编码API和常用注解

三&＃xff0c;详细案例&＃xff08;整合rabbitmq消息队列&＃xff09;

1&＃xff0c;新建父工程 springcloud-tigerhhzz

pom文件代码


xmlns:xsi&＃61;"http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation&＃61;"http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
4.0.0
com.tigerhhzz.springcloud-tigerhhzz
springcloud-tigerhhzz
pom
1.0-SNAPSHOT

cloud-api-commons
cloud-eureka-server7001
cloud-eureka-server7002
cloud-comsumer-order80
cloud-provider-payment8001
cloud-provider-payment8002
cloud-comsumer-feign-order80
cloud-provider-hystrix-payment8001
cloud-comsumer-feign-hystrix-order80
cloud-comsumer-hystrix-dashboard9001
cloud-gateway-gateway9527
cloud-config-center-3344
cloud-config-client-3355
cloud-config-client-3366
cloud-stream-rabbitmq-provider8801
cloud-stream-rabbitmq-consumer8802
cloud-stream-rabbitmq-consumer8803


UTF-8
1.8
1.8
4.12
1.2.17
1.18.0
<mysql.version>5.1.47
1.1.16
1.3.2






org.springframework.boot
spring-boot-dependencies
2.2.2.RELEASE
pom
import



org.springframework.cloud
spring-cloud-dependencies
Hoxton.SR1
pom
import



com.alibaba.cloud
spring-cloud-alibaba-dependencies
2.1.0.RELEASE
pom
import


mysql
mysql-connector-java
$mysql.version


com.alibaba
druid
$druid.version


org.mybatis.spring.boot
mybatis-spring-boot-starter
$mybatis.spring.boot.version


log4j
log4j
$log4j.version


junit
junit
$junit.version


org.projectlombok
lombok
$lombok.version
true






2&＃xff0c;新建注册中心模块&＃xff08;本文使用erueka作为注册中心&＃xff09;

cloud-eureka-server7001
项目结构&＃xff1a;

项目main入口方法&＃xff1a;

package com.tigerhhzz.springcloud;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.netflix.eureka.server.EnableEurekaServer;
/**
* &＃64;author tigerhhzz
* &＃64;date 2022/6/13 10:33
*/
&＃64;SpringBootApplication
&＃64;EnableEurekaServer
public class EurekaMain7001
public static void main(String[] args)
SpringApplication.run(EurekaMain7001.class,args);



pom.xml


xmlns:xsi&＃61;"http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation&＃61;"http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">

springcloud-tigerhhzz
com.tigerhhzz.springcloud-tigerhhzz
1.0-SNAPSHOT

4.0.0
cloud-eureka-server7001


org.springframework.cloud
spring-cloud-starter-netflix-eureka-server


org.springframework.boot
spring-boot-starter-web


org.springframework.boot
spring-boot-starter-actuator


com.tigerhhzz.springcloud-tigerhhzz
cloud-api-commons
1.0-SNAPSHOT





application.yml

server:
port: 7001
spring:
application:
name: cloud-eureka-server7001
eureka:
instance:
hostname: eureka7001.com
client:
register-with-eureka: false # 表示不向注册中心注册
fetch-registry: false # 由于注册中心的职责就是维护服务实例&＃xff0c;所以它不需要去检索服务
service-url:
#defaultZone: http://eureka7002.com:7002/eureka/ #集群 指向其他eureka
defaultZone: http://eureka7001.com:7001/eureka/ #单机 指向自己
server:
enable-self-preservation: false #关闭自我保护机制 &＃xff0c;保证不可用服务被及时剔除


3&＃xff0c;新建三个子模块

cloud-stream-rabbitmq-provider8801&＃xff0c;作为生产者进行消息模块

cloud-stream-rabbitmq-consumer8802&＃xff0c;作为消费者接受模块

cloud-stream-rabbitmq-consumer8803&＃xff0c;作为消费者接受模块

目录结构&＃xff1a;
module:cloud-stream-rabbitmq-provider8801
pom.xml


xmlns:xsi&＃61;"http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation&＃61;"http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">

springcloud-tigerhhzz
com.tigerhhzz.springcloud-tigerhhzz
1.0-SNAPSHOT

4.0.0
cloud-stream-rabbitmq-provider8801


org.springframework.cloud
spring-cloud-starter-stream-rabbit


org.springframework.cloud
spring-cloud-starter-netflix-eureka-client


org.springframework.boot
spring-boot-starter-web


org.springframework.boot
spring-boot-starter-actuator


org.springframework.boot
spring-boot-starter-test
test


com.tigerhhzz.springcloud-tigerhhzz
cloud-api-commons
1.0-SNAPSHOT





application.yml

server:
port: 8801
spring:
application:
name: cloud-stream-privider
cloud:
stream:
binders: #自此处配置要绑定的rabbitmq的服务信息
defaultRabbit: #表示定义的名称&＃xff0c;用于binding整合
type: rabbit #消息组件类型
environment: # 设置rabbitmq的相关的环境配置
spring:
rabbitmq:
host: localhost
port: 5672
username: guest
password: guest
bindings: #服务的整合处理
output: #这个名字是一个通道的名称
destination: studyExchange #表示要使用的exchange名称定义
content-type: application/json #设置消息类型&＃xff0c;本次为json
binder: defaultRabbit #设置要绑定的消息服务的具体设置
eureka:
client:
service-url:
defaultZone: http://eureka7001.com:7001/eureka/ #http://localhost:7001/eureka/
instance:
lease-expiration-duration-in-seconds: 5 #如果现在超过了5秒的间隔
lease-renewal-interval-in-seconds: 2 #设置心跳的时间间隔
instance-id: send-8801.com
prefer-ip-address: true #访问的路径变为IP地址


主启动类

package com.tigerhhzz.springcloud;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
/**
* &＃64;author tigerhhzz
* &＃64;date 2022/6/18 10:00
*/
&＃64;SpringBootApplication
public class StreamMQMain8801
public static void main(String[] args)
SpringApplication.run(StreamMQMain8801.class,args);



业务类–接口

package com.tigerhhzz.springcloud.service;
/**
* &＃64;author tigerhhzz
* &＃64;date 2022/6/18 10:01
*/
public interface IMessageService
public String send();


业务类–实现类

package com.tigerhhzz.springcloud.service.impl;
import cn.hutool.core.lang.UUID;
import com.tigerhhzz.springcloud.service.IMessageService;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.MessageBuilder;
import javax.annotation.Resource;
/**
* &＃64;author tigerhhzz
* &＃64;date 2022/6/18 10:01
*/
&＃64;EnableBinding(Source.class) //定义消息推送通道
public class MessageServiceImpl implements IMessageService
//消息发送通道
&＃64;Resource
private MessageChannel output;
&＃64;Override
public String send()
String serial &＃61; UUID.randomUUID().toString();
output.send(MessageBuilder.withPayload(serial).build());
System.out.println("***********serial:"&＃43;serial);
return serial;



controller层

package com.tigerhhzz.springcloud.controller;
import com.tigerhhzz.springcloud.service.IMessageService;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import javax.annotation.Resource;
/**
* &＃64;author tigerhhzz
* &＃64;date 2022/6/18 10:05
*/
&＃64;RestController
public class SendMessageController
&＃64;Resource
private IMessageService messageService;
&＃64;RequestMapping("/sendMessage")
public String sendMessage()
String send &＃61; messageService.send();
return send;



8802基本和生产者8801的module一样

8802的配置&＃xff1a;

server:
port: 8802
spring:
application:
name: cloud-stream-message-consumer
cloud:
stream:
binders: #自此处配置要绑定的rabbitmq的服务信息
defaultRabbit: #表示定义的名称&＃xff0c;用于binding整合
type: rabbit #消息组件类型
environment: # 设置rabbitmq的相关的环境配置
spring:
rabbitmq:
host: localhost
port: 5672
username: guest
password: guest
bindings: #服务的整合处理
input: #这个名字是一个通道的名称
destination: studyExchange #表示要使用的exchange名称定义
content-type: application/json #设置消息类型&＃xff0c;本次为json
binder: defaultRabbit #设置要绑定的消息服务的具体设置
eureka:
client:
service-url:
defaultZone: http://eureka7001.com:7001/eureka/ #http://localhost:7001/eureka/
instance:
lease-expiration-duration-in-seconds: 5 #如果现在超过了5秒的间隔
lease-renewal-interval-in-seconds: 2 #设置心跳的时间间隔
instance-id: receiver-8802.com
prefer-ip-address: true #访问的路径变为IP地址


8802的控制层&＃xff1a;

package com.tigerhhzz.springcloud.controller;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;
/**
* &＃64;author tigerhhzz
* &＃64;date 2022/6/18 10:32
*/
&＃64;Component
&＃64;EnableBinding(Sink.class)
public class ReceiveMessageListenerController
&＃64;Value("$server.port")
private String serverPort;
&＃64;StreamListener(Sink.INPUT)
public void input(Message message)
System.out.println("我是消费者8802&＃xff0c;-----》接受到的消息是&＃xff1a;"&＃43;message.getPayload()&＃43;"\\t"&＃43;serverPort);



依照8802&＃xff0c;clone出来一份运行8803

4&＃xff0c;测试

依次启动rabbitmq
7001注册中心
8801消息生产模块
8802和8803消息消费模块


eureka7001注册中心

测试&＃xff1a; 8801发送两条消息
http://localhost:8801/sendMessage

***********serial:76f1e833-100b-4d30-9423-4de0fad7e71e
***********serial:cfbcbed8-ba27-4036-8c5b-264a44cffea4


8802和8803分别接收到两条消息

我是消费者8802&＃xff0c;-----》接受到的消息是&＃xff1a;76f1e833-100b-4d30-9423-4de0fad7e71e 8802
我是消费者8802&＃xff0c;-----》接受到的消息是&＃xff1a;cfbcbed8-ba27-4036-8c5b-264a44cffea4 8802


我是消费者8803&＃xff0c;-----》接受到的消息是&＃xff1a;76f1e833-100b-4d30-9423-4de0fad7e71e 8803
我是消费者8803&＃xff0c;-----》接受到的消息是&＃xff1a;cfbcbed8-ba27-4036-8c5b-264a44cffea4 8803


5&＃xff0c;分组消费与持久化

微服务应用放置于同一个group中&＃xff0c;就能够保证消息只会被其中一个应用消费一次。不同的组是可以消费的&＃xff0c;同一个组内会发生竞争关系&＃xff0c;只有其中一个可以消费。

8803/8802都变成相同组&＃xff0c;group两个相同

8802/8803实现了轮询分组&＃xff0c;每次只有一个消费者。8801模块的发的消息只能被8802或者8803其中一个接受到&＃xff0c;这样避免了重复消费

加上分组group属性&＃xff0c;可以实现消息队列的持久化&＃xff0c;group属性是持久化消息队列中很重要的一个属性。

rabiitmq界面&＃xff1a;